package broker

import (
	"fmt"
	"log"
	"net"

	//"strings"
	"sync"
	"time"

	"github.com/yangyongzhen/mqtt-broker/internal/auth"
	"github.com/yangyongzhen/mqtt-broker/internal/protocol/common"
	"github.com/yangyongzhen/mqtt-broker/internal/storage"
	"github.com/yangyongzhen/mqtt-broker/internal/storage/memory"
	redisStore "github.com/yangyongzhen/mqtt-broker/internal/storage/redis"
)

type Broker struct {
	listener     net.Listener
	clients      map[string]*Client
	topicManager *TopicManager
	store        storage.Store
	auth         auth.Authenticator
	mu           sync.RWMutex
	running      bool
	config       *Config
}

type Config struct {
	Server  ServerConfig
	Storage StorageConfig
}

type ServerConfig struct {
	Port           int `json:"port" yaml:"port"`
	MaxConnections int `json:"max_connections" yaml:"max_connections"`
	MaxMessageSize int `json:"max_message_size" yaml:"max_message_size"`
	//RetainedMsgLimit int
	//SessionExpiry    time.Duration
	//MessageExpiry    time.Duration
}

type StorageConfig struct {
	Type  string
	Redis *redisStore.Config
}

func NewBroker(config *Config, authenticator auth.Authenticator) (*Broker, error) {
	var store storage.Store
	var err error
	switch config.Storage.Type {
	case "redis":
		if config.Storage.Redis == nil {
			return nil, fmt.Errorf("redis config is required when storage type is redis")
		}
		store, err = redisStore.NewRedisStore(config.Storage.Redis)
		if err != nil {
			return nil, fmt.Errorf("failed to create redis store: %v", err)
		}
	case "memory":
		store = memory.NewMemoryStore()
	default:
		store = memory.NewMemoryStore()
	}
	return &Broker{
		clients:      make(map[string]*Client),
		topicManager: NewTopicManager(),
		store:        store,
		auth:         authenticator,
		config:       config,
	}, nil
}

func (b *Broker) Start() error {
	address := fmt.Sprintf(":%d", b.config.Server.Port)
	log.Printf("Starting MQTT broker on %s\n", address)
	listener, err := net.Listen("tcp", address)
	if err != nil {
		return err
	}

	b.listener = listener
	b.running = true

	fmt.Printf("MQTT Broker started on %s\n", address)

	// 启动清理任务
	go b.cleanupTask()

	for b.running {
		conn, err := listener.Accept()
		if err != nil {
			if b.running {
				fmt.Printf("Accept error: %v\n", err)
			}
			continue
		}

		// 检查连接数限制
		b.mu.RLock()
		clientCount := len(b.clients)
		b.mu.RUnlock()

		if clientCount >= b.config.Server.MaxConnections {
			fmt.Printf("Max connections reached, rejecting new connection\n")
			conn.Close()
			continue
		}

		client := NewClient(conn, b)
		go client.Start()
	}

	return nil
}

func (b *Broker) Stop() {
	b.running = false
	if b.listener != nil {
		b.listener.Close()
	}

	b.mu.Lock()
	defer b.mu.Unlock()

	for _, client := range b.clients {
		client.Close()
	}
}

func (b *Broker) AddClient(clientID string, client *Client) {
	b.mu.Lock()
	defer b.mu.Unlock()

	// 如果客户端已存在，断开旧连接
	if _, exists := b.clients[clientID]; exists {
		//fmt.Printf("Client %s reconnected, closing old connection\n", clientID)
		//existingClient.Close()
	}

	b.clients[clientID] = client
	fmt.Printf("Client %s connected, total clients: %d\n", clientID, len(b.clients))
}

func (b *Broker) RemoveClient(clientID string) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if _, exists := b.clients[clientID]; exists {
		delete(b.clients, clientID)
		fmt.Printf("Client %s disconnected, total clients: %d\n", clientID, len(b.clients))
	}

	// 清理订阅
	b.topicManager.RemoveClientSubscriptions(clientID)
}

func (b *Broker) PublishMessage(message *common.Message) {
	// 验证消息大小
	if len(message.Payload) > b.config.Server.MaxMessageSize {
		fmt.Printf("Message too large: %d bytes\n", len(message.Payload))
		return
	}

	// 获取订阅者
	subscribers := b.topicManager.GetSubscribers(message.Topic)

	b.mu.RLock()
	defer b.mu.RUnlock()

	// 发送给所有订阅者
	for clientID, subscription := range subscribers {
		if client, exists := b.clients[clientID]; exists {
			// 选择较小的QoS
			qos := message.QoS
			if subscription.QoS < qos {
				qos = subscription.QoS
			}

			client.PublishToClient(message, qos)
		}
	}

	// 处理保留消息
	if message.Retain {
		if len(message.Payload) == 0 {
			// 空载荷删除保留消息
			b.store.DeleteRetainedMessage(message.Topic)
		} else {
			b.store.SaveRetainedMessage(message.Topic, message)
		}
	}

	fmt.Printf("Published message to topic %s, %d subscribers\n", message.Topic, len(subscribers))
}

func (b *Broker) authenticateClient(info *common.ClientInfo) bool {
	if b.auth == nil {
		return true // 无认证器时允许所有连接
	}

	return b.auth.Authenticate(info.Username, info.Password)
}

func (b *Broker) getMatchingRetainedMessages(topicFilter string) []*common.Message {
	var messages []*common.Message

	// 这里需要实现获取所有保留消息并过滤的逻辑
	// 由于存储接口限制，这里简化实现
	if msg, err := b.store.LoadRetainedMessage(topicFilter); err == nil && msg != nil {
		messages = append(messages, msg)
	}

	return messages
}

func (b *Broker) cleanupTask() {
	ticker := time.NewTicker(1 * time.Hour)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			b.cleanupExpiredSessions()
			b.cleanupExpiredMessages()
		}
	}
}

func (b *Broker) cleanupExpiredSessions() {
	// 清理过期会话的逻辑
	// 这里需要扩展存储接口来支持会话遍历
	fmt.Println("Cleaning up expired sessions...")
}

func (b *Broker) cleanupExpiredMessages() {
	// 清理过期消息的逻辑
	fmt.Println("Cleaning up expired messages...")
}

func (b *Broker) GetStats() map[string]interface{} {
	b.mu.RLock()
	defer b.mu.RUnlock()

	return map[string]interface{}{
		"connected_clients": len(b.clients),
		"max_connections":   b.config.Server.MaxConnections,
		"uptime":            time.Since(time.Now()), // 需要记录启动时间
	}
}
